package com.atguigu.flink.datastream.transform;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 14:06
 * <p>
 * 简单转换算子:
 * map:     映射， 一对一
 * filter:  过滤
 * flatMap: 扁平映射
 */
public class Flink01_SimpleTransform {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        ds.print("input");

        //map
        SingleOutputStreamOperator<String> map = ds.map(
                //event -> JSON.toJSONString(event)
                JSON::toJSONString

                /*
                new MapFunction<Event, String>() {
                    @Override
                    public String map(Event value) throws Exception {
                        return JSON.toJSONString(value);
                    }
                }

                 */
        );
        //map.print();

        //filter
        SingleOutputStreamOperator<Event> filter = ds.filter(
                new FilterFunction<Event>() {
                    @Override
                    public boolean filter(Event value) throws Exception {
                        return "Zhangs".equals(value.getUser()) || "Lisi".equals(value.getUser());
                    }
                }
        );
        //filter.print("filter");

        //flatMap
        SingleOutputStreamOperator<String> flatMap = ds.flatMap(
                new FlatMapFunction<Event, String>() {
                    @Override
                    public void flatMap(Event value, Collector<String> out) throws Exception {
                        if ("Zhangs".equals(value.getUser())) {
                            out.collect(value.getUser());
                            out.collect(value.getUrl());
                            out.collect(value.getTs().toString());
                        }
                    }
                }
        );

        flatMap.print("flatMap");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
